package hbase;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableWrapper;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * Created by spark on 16-10-18.
 */
public class RegionObserveExample extends BaseRegionObserver{

    private static final Log LOG = LogFactory.getLog(RegionObserveExample.class);


    private static final byte[] FIXED_ROW = Bytes.toBytes("@@@GETTIME@@@");

    private HTableWrapper indexTable;

    private static final byte[] FAMILY = Bytes.toBytes("data");

    private static final byte[] QUALIFIER = Bytes.toBytes("content");



    @Override
    public void start(CoprocessorEnvironment e) throws IOException {
        indexTable = (HTableWrapper) e.getTable(TableName.valueOf("indexTable"));
        LOG.info("start region observer example");
    }

    @Override
    public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, List<Cell> results) throws IOException {
        if (Bytes.equals(get.getRow(), FIXED_ROW)) {
            KeyValue kv = new KeyValue(get.getRow(), FIXED_ROW, FIXED_ROW, 1,
                    Bytes.toBytes(System.currentTimeMillis()));
            results.add(kv);
            e.bypass();
        }
    }

    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        LOG.info("begin create index for " + Bytes.toString(put.getRow()));

        List<Cell> cells = put.get(FAMILY, QUALIFIER);

//        Connection conn = ConnectionFactory.createConnection(e.getEnvironment().getConfiguration());
//        Table indexTable = conn.getTable(TableName.valueOf("indexTable"));

        for (Cell cell:
             cells) {
            Put indexPut = new Put(CellUtil.cloneValue(cell));
            indexPut.addColumn(FAMILY, QUALIFIER, CellUtil.cloneRow(cell));
            indexTable.put(indexPut);
        }
        indexTable.close();
        LOG.info("create index end!");
    }
}
